package com.atguigu.edu.realtime.function;

import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema;
import org.apache.kafka.clients.consumer.ConsumerRecord;

/**
 * @ClassName: KafkaDeserWithFilter
 * @Description: TODO kafka消费者反序列化类,过滤脏数据
 * @Author: zhaoxunfeng
 * @Date: 2022-08-31 12:23
 * @Version: 1.0.0
 */
public class KafkaDeserWithFilter implements KafkaDeserializationSchema<String> {
    @Override
    public boolean isEndOfStream(String s) {
        return false;
    }

    @Override
    public String deserialize(ConsumerRecord<byte[], byte[]> record) throws Exception {

        if (record != null && record.value() != null) {
            return new String(record.value());
        }
        return null;
    }

    @Override
    public TypeInformation<String> getProducedType() {
        return BasicTypeInfo.STRING_TYPE_INFO;
    }
}
